﻿using System;
using System.Collections.Generic;
using System.Linq;
using Apache.Cassandra;

namespace Sandy
{
    public class GetRangeSlicesQuery
    {
        private static IDictionary<string, IDictionary<string, string>> PerformInternal(SandyClient client,
            SandyFamily family, KeyRange keyRange, ColumnParent parent, SlicePredicate pred)
        {
            var res = client.Client.get_range_slices(client.Keyspace, parent, pred, keyRange.ToCassandraRange(),
                ConsistencyLevel.ONE);

            // Prepare result dict
            var dict = new OrderPreservingDictionary<string, IDictionary<string, string>>();
            foreach (var keySlice in res)
            {
                var columnDict = new OrderPreservingDictionary<string, string>();
                foreach (var column in keySlice.Columns)
                {
                    columnDict.Add(family.ColumnNameToString(column.Column.Name),
                                   family.GetValueString(column.Column.Value));
                }
                dict.Add(keySlice.Key, columnDict);
            }

            return dict;
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
                            SandyFamily family, KeyRange keyRange, IEnumerable<string> cols)
        {
            var parent = new ColumnParent { Column_family = family.Name };
            var pred = new SlicePredicate { Column_names = cols.Select(family.ColumnNameToBytes).ToList() };

            return PerformInternal(client, family, keyRange, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
                    SandyFamily family, KeyRange keyRange, Range colRange)
        {
            var parent = new ColumnParent { Column_family = family.Name };
            var pred = new SlicePredicate { Slice_range = colRange.ToSliceRange(family) };

            return PerformInternal(client, family, keyRange, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
            SandyFamily family, KeyRange keyRange, string superCol, IEnumerable<string> cols)
        {
            var parent = new ColumnParent
            {
                Column_family = family.Name,
                Super_column = family.SuperColumnNameToBytes(superCol)
            };

            var pred = new SlicePredicate { Column_names = cols.Select(family.ColumnNameToBytes).ToList() };

            return PerformInternal(client, family, keyRange, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, string>> Perform(SandyClient client,
            SandyFamily family, KeyRange keyRange, string superCol, Range colRange)
        {
            var parent = new ColumnParent
            {
                Column_family = family.Name,
                Super_column = family.SuperColumnNameToBytes(superCol)
            };

            var pred = new SlicePredicate { Slice_range = colRange.ToSliceRange(family) };

            return PerformInternal(client, family, keyRange, parent, pred);
        }

        public static IDictionary<string, IDictionary<string, string>> PerformGetAll(SandyClient client,
            SandyFamily family, KeyRange keyRange, string superCol)
        {
            var parent = new ColumnParent
            {
                Column_family = family.Name,
            };

            var pred = new SlicePredicate
            {
                Column_names = new List<byte[]> { family.SuperColumnNameToBytes(superCol) }
            };

            var res = client.Client.get_range_slices(client.Keyspace, parent, pred, 
                keyRange.ToCassandraRange(), ConsistencyLevel.ONE);

            // Prepare our format
            var dict = new OrderPreservingDictionary<string, IDictionary<string, string>>();
            foreach (var keyslice in res)
            {
                var colOrSuper = keyslice.Columns.FirstOrDefault();
                if (colOrSuper != null)
                    dict.Add(keyslice.Key, family.CreateSupercolDict(colOrSuper.Super_column));
            }

            return dict;
        }

        public static IDictionary<string, string> Perform(SandyClient client,
            SandyFamily family, KeyRange keyRange, string superCol, string column)
        {
            var parent = new ColumnParent
                             {
                                 Column_family = family.Name,
                                 Super_column = family.SuperColumnNameToBytes(superCol)
                             };

            var pred = new SlicePredicate
                           {
                               Column_names = new List<byte[]>() { family.ColumnNameToBytes(column) }
                           };


            var res = client.Client.get_range_slices(client.Keyspace, parent, pred, keyRange.ToCassandraRange(),
                ConsistencyLevel.ONE);

            // Prepare result dict (k -> columnValue)
            var dict = new OrderPreservingDictionary<string, string>();
            foreach (var keySlice in res)
            {
                var resultSuperCol = keySlice.Columns.FirstOrDefault();
                if (resultSuperCol == null) continue;
                var resultCol = resultSuperCol.Column;
                if (resultCol == null) continue;

                dict.Add(keySlice.Key, family.GetValueString(resultCol.Value));
            }

            return dict;
        }


        public static IDictionary<string, string> Perform(SandyClient client,
            SandyFamily family, KeyRange keyRange, string col)
        {
            var parent = new ColumnParent
                             {
                                 Column_family = family.Name
                             };

            var pred = new SlicePredicate
                           {
                               Column_names = new List<byte[]>() { family.ColumnNameToBytes(col) }
                           };

            var res = client.Client.get_range_slices(client.Keyspace, parent, pred, keyRange.ToCassandraRange(),
                ConsistencyLevel.ONE);

            // Prepare result dict (k -> columnValue)
            var dict = new OrderPreservingDictionary<string, string>();
            foreach (var keySlice in res)
            {
                var resultCol = keySlice.Columns.FirstOrDefault();
                if (resultCol == null) continue;

                dict.Add(keySlice.Key, family.GetValueString(resultCol.Column.Value));
            }

            return dict;
        }

        public static IDictionary<string, IDictionary<string, IDictionary<string, string>>>
            PerformSuper(SandyClient client, SandyFamily family, KeyRange keyRange, IEnumerable<string> superCols)
        {
            var pred = new SlicePredicate
            {
                Column_names = superCols.Select(family.SuperColumnNameToBytes).ToList()
            };

            return PerformSuperInternal(client, family, keyRange, pred);
        }

        public static IDictionary<string, IDictionary<string, IDictionary<string, string>>>
            PerformSuper(SandyClient client, SandyFamily family, KeyRange keyRange, Range scolRange)
        {
            var pred = new SlicePredicate
            {
                Slice_range = scolRange.ToSuperSliceRange(family)
            };

            return PerformSuperInternal(client, family, keyRange, pred);
        }

        private static IDictionary<string, IDictionary<string, IDictionary<string, string>>>
            PerformSuperInternal(SandyClient client, SandyFamily family, KeyRange range, SlicePredicate pred)
        {
            var parent = new ColumnParent { Column_family = family.Name };
            var res = client.Client.get_range_slices(client.Keyspace, parent, pred, range.ToCassandraRange(),
                ConsistencyLevel.ONE);

            // Prepare our result dict (k -> (scn -> (cn -> cv)))
            var dict = new OrderPreservingDictionary<string, IDictionary<string, IDictionary<string, string>>>();
            foreach (var keySlice in res)
            {
                var scDict = new OrderPreservingDictionary<string, IDictionary<string, string>>();
                foreach (var scol in keySlice.Columns.Select(c => c.Super_column))
                {
                    scDict.Add(family.SuperColumnNameToString(scol.Name), family.CreateSupercolDict(scol));
                }
                dict.Add(keySlice.Key, scDict);
            }

            return dict;
        }

    }
}